package com.sunday.common.mq.rabbit.study.base.e3_PublishSubscribe;

import com.rabbitmq.client.*;
import com.sunday.common.mq.rabbit.study.base.ConnectionUtil;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;

/**
 * 发布/订阅
 * <p>
 * 在前面的教程中，我们创建了一个作品 队列。工作队列背后的假设是每个任务都是 只交付给一名工人。在这一部分中，我们将做一些事情 完全不同的 - 我们将向多个传递消息 消费者。此模式称为“发布/订阅”。
 * 为了说明这种模式，我们将构建一个简单的日志记录 系统。它将由两个程序组成 - 第一个将发出日志 消息，第二个将接收并打印它们。
 * 在我们的日志记录系统中，接收器程序的每个运行副本都将 获取消息。这样我们就能够运行一个接收器和 将日志定向到磁盘;同时我们将能够运行 另一个接收器，并在屏幕上查看日志。
 * 本质上，已发布的日志消息将广播给所有人 接收器。
 * <p>
 * RabbitMQ中消息传递模型的核心思想是生产者永远不会将任何消息直接发送到队列。实际上，生产者通常甚至不知道消息是否会被传递到任何队列。
 * <p>
 * 相反，生产者只能向交易所发送消息。交换是一件非常简单的事情。一方面，它接收来自生产者的消息，另一方面将它们推送到队列。交易所必须确切知道如何处理收到的消息。
 * 它应该附加到特定队列吗？它应该附加到许多队列吗？或者它应该被丢弃。其规则由交换类型定义 。
 * <p>
 * FANOUT("fanout"),	发布/订阅（广播模式）
 * <p>
 * 不处理路由键。你只需要简单的将队列绑定到交换机上。
 * 一个发送到交换机的消息都会被转发到与该交换机绑定的所有队列上。
 * 很像子网广播，每台子网内的主机都获得了一份复制的消息。Fanout交换机转发消息是最快的。
 */
@Slf4j
public class ReceiveLogsFanout {
    private static final String EXCHANGE_NAME = "fanout_logs";

    public static void main(String[] argv) throws Exception {

        ConnectionFactory factory = ConnectionUtil.factory();

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        /**
         * 有几种可用的交换类型：direct(直接)、topic(主题)、 headers(标头)和 fanout(扇出)。
         * 我们将专注于最后一个 - 扇出。让我们创建 这种类型的交换，并调用它日志：
         *
         * default Exchange不能进行Binding,也不需要进行绑定。
         * 除default Exchange之外，其他任何Exchange都需要和Queue进行Binding，否则无法进行消息路由（转发）
         * Binding的时候，可以设置一个或多个参数，其中参数要特别注意参数类型，如果Routing key中指定的参数类型和消息中指定的参数类型不一致（header Exchange）也不能进行消息转发。
         * Direct Exchange，Topic Exchange进行Binding的时候，需要指定Routing key
         * Fanout Exchange，Headers Exchange进行Binding的时候，不需要指定Routing key。
         */
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);

        push(channel, "A");
        push(channel, "B");

    }


    private static void push(Channel channel, String mark) throws IOException {

        /**
         * 在 Java 客户端中，当我们不向 queueDeclare（） 提供任何参数时，我们会创建一个非持久、独占的、自动删除的队列，并生成一个
         */
        String queueName = channel.queueDeclare().getQueue();
        channel.queueBind(queueName, EXCHANGE_NAME, "");

        log.info(" [*] Waiting for messages. To exit press CTRL+C");

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            log.info(" [x] mark '{}'", mark);
            log.info(" [x] Received '{}'", message);
        };

        channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {
        });

    }

}

